package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.Func.AsyncDimFunction_10;
import com.atguigu.Func.MyWindowFunction;
import com.atguigu.Util.ClickHouseUtil;
import com.atguigu.Util.MyKafkaUtil;
import com.atguigu.bean.TradeProvinceOrderWindow;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @author hjy
 * @create 2023/3/21 8:51
 */
public class Dws10_TradeProvinceOrderWindow {
    public static void main(String[] args) throws Exception {
        //todo 1 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall-flink/check");
//        env.getCheckpointConfig().setCheckpointTimeout(60000L);
//        env.setStateBackend(new HashMapStateBackend());
//        System.setProperty("HADOOP_USER_NAME","atguigu");
        //todo 2 从kafka(dwd_trade_order_detail)读数据
        String topic ="dwd_trade_order_detail";
        String groupID="dws_trade_province_order_window";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupID));
        //todo 3 过滤数据并转为javaBean对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    out.collect(JSON.parseObject(value));
                }
            }
        });
        //todo 4 生成waterMark
        SingleOutputStreamOperator<JSONObject> JSONObjWithWMDS = jsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(TimestampAssignerSupplier.of(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("create_time");
                    }
                })));
        //todo 5 按照order_id 分组
        KeyedStream<JSONObject, String> keyedByIdDS = JSONObjWithWMDS.keyBy(value -> value.getString("id"));
        //todo 6 转为javaBean对象并去重 只要左表数据
        SingleOutputStreamOperator<TradeProvinceOrderWindow> tradeProvinceOrderWindowDS = keyedByIdDS.flatMap(new RichFlatMapFunction<JSONObject, TradeProvinceOrderWindow>() {
            private ValueState<String> existState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //只给5s的乱序
                StateTtlConfig ttlCfg = StateTtlConfig.newBuilder(Time.seconds(5))
                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                        .build();
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("exist_state", String.class);
                valueStateDescriptor.enableTimeToLive(ttlCfg);
                existState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void flatMap(JSONObject value, Collector<TradeProvinceOrderWindow> out) throws Exception {
                String exist = existState.value();
                if (exist == null) {
                    existState.update("1");
                    out.collect(TradeProvinceOrderWindow.builder()
                            .orderId(value.getString("order_id"))
                            .provinceId(value.getString("province_id"))
                            .orderAmount(value.getBigDecimal("split_total_amount"))
                            .build());
                }
            }
        });
        //todo 6.1 按照order_id分组，去重
        SingleOutputStreamOperator<TradeProvinceOrderWindow> tradeProvinceOrderWindowWithOrderID = tradeProvinceOrderWindowDS.map(new RichMapFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow>() {
            private ValueState<String> orderIdState;

            @Override
            public void open(Configuration parameters) throws Exception {
                StateTtlConfig ttlCfg = StateTtlConfig.newBuilder(Time.seconds(5))
                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                        .build();
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("order_id", String.class);

                valueStateDescriptor.enableTimeToLive(ttlCfg);

                orderIdState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public TradeProvinceOrderWindow map(TradeProvinceOrderWindow value) throws Exception {
                String orderExist = orderIdState.value();
                if (orderExist == null) {
                    orderIdState.update("1");
                    value.setOrderCount(1L);
                } else {
                    value.setOrderCount(0L);
                }
                return value;
            }
        });
        //todo 7 按照省份id分组开窗聚合
        SingleOutputStreamOperator<TradeProvinceOrderWindow> reduceDS = tradeProvinceOrderWindowWithOrderID.keyBy(TradeProvinceOrderWindow::getProvinceId)
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow value1, TradeProvinceOrderWindow value2) throws Exception {
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        return value1;
                    }
                }, new WindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeProvinceOrderWindow> input, Collector<TradeProvinceOrderWindow> out) throws Exception {
                        TradeProvinceOrderWindow next = input.iterator().next();
                        out.collect(MyWindowFunction.getJavabeanFields(next, window));
                    }
                });

        //todo 8 关联维表 关联省份表
        //从phoenix读取数据并做旁路缓存 完成--->DimUtil_10
        //异步io，节省等待时间
        SingleOutputStreamOperator<TradeProvinceOrderWindow> tradeProvinceOrderWindowWithProvinceName = AsyncDataStream.unorderedWait(reduceDS, new AsyncDimFunction_10<TradeProvinceOrderWindow>("DIM_BASE_PROVINCE") {
            @Override
            public void joinDimInfo(TradeProvinceOrderWindow input, JSONObject dimInfo) {

                input.setProvinceName(dimInfo.getString("NAME"));
            }

            @Override
            public String getId(TradeProvinceOrderWindow input) {
                return input.getProvinceId();
            }
        }, 60, TimeUnit.SECONDS);

        //todo 9 写入clickHouse
        tradeProvinceOrderWindowWithProvinceName.print("tradeProvinceOrderWindowWithProvinceName>>>>>>>>>>>>>>");
        tradeProvinceOrderWindowWithProvinceName.addSink(ClickHouseUtil.getSinkFunction("insert into dws_trade_province_order_window values(?,?,?,?,?,?,?)"));

        //todo 10 启动程序
        env.execute();

    }
}
